package com.xiaoyg.work;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * @author xiaoyg
 * @Title: Work
 * @ProjectName rabbitmq-task
 * @Description: TODO
 * @date 2018/12/317:03
 */
public class Work {

    //队列名称
    private final static String TASK_QUEUE_NAME = "task_queue_name";

    public static void main(String[] args) {
        //区分不同工作进程的输出
        int hashCode = Work.class.hashCode();
        //创建连接到RabbitMQ
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置RabbitMQ所在主机的IP或者主机名
        connectionFactory.setHost("192.168.8.129");
        //制定用户和密码
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("123456");
        //指定端口
        connectionFactory.setPort(AMQP.PROTOCOL.PORT);
        //创建一个连接
        try {
            Connection connection = connectionFactory.newConnection();
            //创建一个频道
            Channel channel = connection.createChannel();

            //设置消息持久化  RabbitMQ不允许使用不同的参数重新定义一个队列，所以已经存在的队列，我们无法修改其属性。
            boolean durable = true;
            //指定一个队列
            channel.queueDeclare(TASK_QUEUE_NAME,durable,false,false,null);

            //创建队列消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);

            /**
             * ack= true: Round-robin 转发   消费者被杀死，消息会丢失
             * ack=false:消息应答 ，为了保证消息永远不会丢失，RabbitMQ支持消息应答（message acknowledgments）。
             * 消费者发送应答给RabbitMQ，告诉它信息已经被接收和处理，然后RabbitMQ可以自由的进行信息删除。
             * 如果消费者被杀死而没有发送应答，RabbitMQ会认为该信息没有被完全的处理，然后将会重新转发给别的消费者。
             * 通过这种方式，你可以确认信息不会被丢失，即使消者偶尔被杀死。
             * 消费者需要耗费特别特别长的时间是允许的。
             *
             */

            //打开应答机制
            boolean ack = false;

            //指定消费队列
            channel.basicConsume(TASK_QUEUE_NAME,ack,consumer);

            int prefetchCount = 1;
            channel.basicQos(prefetchCount);

            while (true){
                //nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(hashCode+"Received Message:'"+message+"'");
                doWork(message);
                System.out.println(hashCode+"Received Done");
                //发送应答
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            }

        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 每个点耗时1s
     * @param task
     * @throws InterruptedException
     */
    private static void doWork(String task) throws InterruptedException {
        for (char ch: task.toCharArray()) {
            if(ch == '.'){
                Thread.sleep(1000);
            }
        }
    }

}
